package demo.canal;

import com.alibaba.fastjson.JSONObject;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData;
import com.github.shyiko.mysql.binlog.event.EventData;
import com.github.shyiko.mysql.binlog.event.TableMapEventData;
import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData;
import com.github.shyiko.mysql.binlog.event.WriteRowsEventData;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import com.github.shyiko.mysql.binlog.network.SSLMode;

import java.io.IOException;
import java.io.Serializable;
import java.util.Map;

/**
 * 使用 mysql-binlog-connector 组件监听binlog
 */
public class MyLogBinClient {

    public static void main(String[] args) {
        BinaryLogClient logClient = new BinaryLogClient("localhost", 3306, "root", "123456");
        EventDeserializer eventDeserializer = new EventDeserializer();
        eventDeserializer.setCompatibilityMode(EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY,
                                               EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG);
        logClient.setSSLMode(SSLMode.REQUIRED);

        logClient.setEventDeserializer(eventDeserializer);
        logClient.registerLifecycleListener(new BinaryLogClient.LifecycleListener() {
            @Override
            public void onConnect(BinaryLogClient client) {
                System.out.println("启动完成");
            }

            @Override
            public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
                System.out.println("监听binlog异常");
            }

            @Override
            public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
                System.out.println("事件序列化失败");
            }

            @Override
            public void onDisconnect(BinaryLogClient client) {
                System.out.println("客户端断开连接");
            }
        });
        logClient.registerEventListener(event -> {
            EventData eventData = event.getData();
            if (eventData instanceof TableMapEventData) {
                TableMapEventData tableMapEventData = (TableMapEventData) eventData;
                System.out.println(String.format("表Id：%s，数据库：%s，表名：%s", tableMapEventData.getTableId(), tableMapEventData.getDatabase(),
                                                 tableMapEventData.getTable()));
            }
            if (eventData instanceof UpdateRowsEventData) {
                System.out.println("修改数据：" + eventData);
                UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) eventData;
                for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
                    Serializable[] rowKey = row.getKey();
                    for (Serializable serializable : rowKey) {
                        if (serializable instanceof byte[]) {
                            System.out.println(new String((byte[]) serializable).trim());
                        } else {
                            System.out.println(serializable);
                        }
                    }
                }
            }
            if (eventData instanceof WriteRowsEventData) {
                System.out.println("插入数据：" + eventData);
            }
            if (eventData instanceof DeleteRowsEventData) {
                System.out.println("删除：" + eventData);
            }
        });

        try {

            logClient.connect();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
